package org.cainiao.basic.reactive;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

public class Test {

    public static void main(String[] args) {
        test();
    }
    
    public static void test3() {
        Flux<Integer> ints = Flux.range(1, 3);
        ints.subscribe();
    }

    public static void test4() {
        Flux<Integer> ints = Flux.range(1, 3);
        ints.subscribe(System.out::println);
    }

    public static void test5() {
        Flux<Integer> ints = Flux.range(1, 4)
            .map(i -> {
                if (i <= 3) {
                    return i;
                }
                throw new RuntimeException("Got to 4");
            });
        ints.subscribe(
            System.out::println,
            error -> System.err.println("Error: " + error));
    }

    public static void test6() {
        Flux<Integer> ints = Flux.range(1, 4);
        ints.subscribe(System.out::println,
            error -> System.err.println("Error " + error),
            () -> System.out.println("Done"));
    }

    public static void test7() {
        SampleSubscriber<Integer> ss = new SampleSubscriber<>();
        Flux<Integer> ints = Flux.range(1, 4);
        ints.subscribe(ss);
    }

    public static void test() {
        Flux.range(1, 10)
            .doOnRequest(r -> System.out.println("request of " + r))
            .subscribe(new BaseSubscriber<>() {
                @Override
                public void hookOnSubscribe(Subscription subscription) {
                    request(1);
                }

                @Override
                public void hookOnNext(Integer integer) {
                    System.out.println("Cancelling after having received " + integer);
                    cancel();
                }
            });
    }

}
